package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/1/30
 *
 *      计算size = 3的基于元素个数的滚动窗口，统计每种传感器的vc的和。
 *
 *     无keyBy，不聚合
 *
 *
 */
public class Demo9_GetTimeAttr
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                   /*
                        reduce(
                            ReduceFunction<T> reduceFunction,
                             WindowFunction<T, R, K, W> function
                           )
                    */
                   .reduce(new ReduceFunction<WaterSensor>()
                           {
                               @Override
                               public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                                   System.out.println("Demo7_Reduce.reduce");
                                   value2.setVc(value1.getVc() + value2.getVc());
                                   return value2;
                               }
                           },
                       /*
                            IN： ReduceFunction的输出。
                                        ReduceFunction 聚合后的最终结果会输出到WindowFunction
                             OUT: 最终的输出
                              KEY: key
                               W : 窗口。通过这个参数获取时间窗口中时间属性
                        */
                       new WindowFunction<WaterSensor, String, String, TimeWindow>()
                       {
                           /*
                                apply: 非滚动聚合。窗口到点了被计算了只运行一次。

                                 Iterable<WaterSensor> input:  只有一条数据!

                            */
                           @Override
                           public void apply(String key, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {

                               WaterSensor result = input.iterator().next();
                               out.collect(key +":在这个时间 "+ MyUtil.parseTime(window) +"vc之和是:" + result.getVc());
                           }
                       })
                   .print();

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
